package com.yxc.producer.thread;

import com.yxc.producer.handler.KafkaSendResultHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * @author: YXC
 * @time: 2022/12/7 9:55
 **/
@Slf4j
@Component
public class KafkaProducer implements ApplicationRunner {

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;
    @Autowired
    private KafkaSendResultHandler kafkaSendResultHandler;

    public static final String TOPIC = "kafka-demo-topic";

    @Override
    public void run(ApplicationArguments args) throws Exception {
        kafkaTemplate.setProducerListener(kafkaSendResultHandler);
        log.info("开始发送消息");
        new Thread(() -> {
            while (true) {
                kafkaTemplate.send(TOPIC, "message: " + System.currentTimeMillis());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).run();
    }
}
